package aos2;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;

public class MyServer {
	private int id;
	private int listenPort;
	private boolean stop = false;
	private Client client;
	private int logicalClock;
	private ConcurrentHashMap<String, LinkedBlockingQueue<TailServerRequest>> localQMap = new ConcurrentHashMap<String, LinkedBlockingQueue<TailServerRequest>>();
	private ConcurrentHashMap<String, HashMap<String, ServerRequest>> localReqHistoryMap = new ConcurrentHashMap<String, HashMap<String, ServerRequest>>();
	private Hashtable<String, Integer> neighborServers = new Hashtable<String, Integer>();
	private int seqnum;
	private static int offset;
	private ConcurrentHashMap<String, HashMap<String, MyVector>> mapMap = new ConcurrentHashMap<String, HashMap<String, MyVector>>();
	private ConcurrentHashMap<String, PriorityBlockingQueue<MyVector>> queueMap = new ConcurrentHashMap<String, PriorityBlockingQueue<MyVector>>();
	private boolean isTail = false;
	private String tailIP;
	private int tailPort;

	public MyServer(String configFile) {
		client = new Client();

		Properties prop = new Properties();
		Properties neighbors = new Properties();
		try {
			prop.load(new FileInputStream(new File(configFile)));
			neighbors.load(new FileInputStream(new File("servers.properties")));
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}

		this.id = Integer.parseInt(prop.getProperty("ID"));
		this.listenPort = Integer.parseInt(prop.getProperty("PORT"));
		int tailid = Integer.parseInt(prop.getProperty("TAIL_ID"));
		this.tailIP = prop.getProperty("TAIL_IP");
		this.tailPort = Integer.parseInt(prop.getProperty("TAIL_PORT"));
		if (id == tailid) {
			isTail = true;
		}
		offset = Integer.parseInt(prop.getProperty("OFFSET"));

		for (Entry<Object, Object> es : neighbors.entrySet()) {
			neighborServers.put((String) es.getKey(),
					Integer.parseInt((String) es.getValue()));
		}
	}

	public boolean startServer() {

		Runnable r1 = new Runnable() {
			public void run() {
				try {
					final ServerSocket socket = new ServerSocket(listenPort);

					while (!stop) {
						final Socket skt = socket.accept();
						Runnable r = new Runnable() {
							public void run() {
								handleRequest(skt);
							}
						};
						Thread t = new Thread(r);
						t.start();
					}
				} catch (IOException e) {

				}
			}
		};
		Thread t1 = new Thread(r1);
		t1.start();

		if (!isTail) {
			Runnable r2 = new Runnable() {
				public void run() {
					while (!stop) {
						synchronized (this) {
						//	synchronized (localQMap) {
							Set<Entry<String, LinkedBlockingQueue<TailServerRequest>>> es = localQMap
									.entrySet();
							for (Entry<String, LinkedBlockingQueue<TailServerRequest>> e : es) {
								
									while (e.getValue().size() != 0) {
										ServerRequest sr = e.getValue().poll();
										System.out.println("make req to tail:"
												+ sr.getMessage());
										client.makeRequest(tailIP, tailPort, sr);
									}
					//			}
							}
						}
					}
				}
			};
			Thread t2 = new Thread(r2);
			t2.start();
		}
		if (isTail) {
			Runnable r3 = new Runnable() {
				public void run() {
					while (!stop) {
						synchronized (this) {
							Set<Entry<String, PriorityBlockingQueue<MyVector>>> es = queueMap
									.entrySet();
							for (Entry<String, PriorityBlockingQueue<MyVector>> e : es) {
								while (e.getValue().size() > 4) {
									MyVector rsp = e.getValue().poll();
									String response = rsp.key;
									System.out.println(rsp.msg);
									Request r = new Request();
									r.setMessage(rsp.msg);
									r.setFileName(rsp.file);
									writeFile(r);
									r.setMessage(response);
									r.setType(RequestType.TAILSERVER_WRITE_RESPONSE);
									client.makeRequest("localhost",
											listenPort - 1, r);
									mapMap.get(rsp.file).remove(response);
								}
							}
							// try {
							// Thread.sleep(100);
							// } catch (InterruptedException e) {
							// e.printStackTrace();
							// }
						}
					}
				}
			};
			Thread t3 = new Thread(r3);
			t3.start();
		}
		return true;
	}

	protected void handleRequest(Socket skt) {
		try {
			InputStream is = skt.getInputStream();
			BufferedReader br = new BufferedReader(new InputStreamReader(is));
			String incoming = null;
			OutputStream os = skt.getOutputStream();
			PrintWriter pw = new PrintWriter(os, true);
			while ((incoming = br.readLine()) != null) {
				if (incoming.equals("Bye.")) {
					break;
				}
				Request req = parseRequest(incoming);
				switch (req.getType()) {
				case CLIENT_READ:
					if (isTail) {
						String line = readFile(req.getFileName());
						pw.println(line);
						pw.println("Bye.");
					}
					if (!isTail) {
						if (!localReqHistoryMap.containsKey(req.getFileName())) {
							localReqHistoryMap.put(req.getFileName(),
									new HashMap<String, ServerRequest>());
						}
						if (localReqHistoryMap.get(req.getFileName()).isEmpty()) {
							String line = readFile(req.getFileName());
							pw.println(line);
							pw.println("Bye.");
						} else {
							req.setType(RequestType.TAILSERVER_READ);
							String reply = client.makeRequest(tailIP, tailPort,
									req);
							pw.println(reply);
							pw.println("Bye.");
						}
					}
					break;
				case CLIENT_WRITE:
					synchronized (this) {
						logicalClock += offset;
						seqnum++;
					}
					ServerRequest curReq = new ServerRequest();
					curReq.setFileName(req.getFileName());
					curReq.setMessage(req.getMessage());
					curReq.setSeqNum(seqnum);
					curReq.setTimeStamp(logicalClock);
					curReq.setType(RequestType.SERVER_WRITE);
					curReq.setRequestorId(id);
					pw.println("Bye.");
					if (localReqHistoryMap.containsKey(curReq.getFileName())) {
						localReqHistoryMap.get(curReq.getFileName()).put(
								id + "-" + seqnum, curReq);
					} else {
						localReqHistoryMap.put(req.getFileName(),
								new HashMap<String, ServerRequest>());
						localReqHistoryMap.get(curReq.getFileName()).put(
								id + "-" + seqnum, curReq);
					}
					/*
					 * for (Entry<String, Integer> es :
					 * neighborServers.entrySet()) {
					 * System.out.println("fwd req to " + es.getKey() + ":" +
					 * es.getValue()); client.makeRequest(es.getKey(),
					 * es.getValue(), curReq); }
					 */
					for (int i = 1; i <= 3; i++) {
						if (i != id)
							client.makeRequest("localhost", 5000 + (i - 1),
									curReq);
					}

					break;
				case SERVER_WRITE:
					ServerRequest sreq = null;
					synchronized (this) {// wait-notify
						logicalClock += offset;
						sreq = parseServerRequest(incoming);
						TailServerRequest tsreq = getTailServerRequest(sreq);
						if (localQMap.containsKey(tsreq.getFileName())) {
							localQMap.get(tsreq.getFileName()).add(tsreq);
						} else {
							localQMap
									.put(tsreq.getFileName(),
											new LinkedBlockingQueue<TailServerRequest>());
							localQMap.get(tsreq.getFileName()).add(tsreq);
						}
					}
					// add to local q
					pw.println("Bye.");
					if (localReqHistoryMap.containsKey(req.getFileName())) {
						localReqHistoryMap.get(req.getFileName()).put(
								sreq.getRequestorId() + "-" + sreq.getSeqNum(),
								sreq);
					} else {
						localReqHistoryMap.put(req.getFileName(),
								new HashMap<String, ServerRequest>());
						localReqHistoryMap.get(req.getFileName()).put(
								sreq.getRequestorId() + "-" + sreq.getSeqNum(),
								sreq);
					}
					break;
				case TAILSERVER_WRITE:
					synchronized (this) {
						logicalClock += offset;
					}
					TailServerRequest tsreq = parseTailServerRequest(incoming);
					String key = tsreq.getRequestorId() + "-"
							+ tsreq.getSeqNum();// this forms a unique key
					if (!mapMap.containsKey(tsreq.getFileName())) {
						mapMap.put(tsreq.getFileName(),
								new HashMap<String, MyVector>());
					}
					if (!mapMap.get(tsreq.getFileName()).containsKey(key)) {
						System.out.println("Map dont have key");
						MyVector myv = new MyVector(4);// neighborServers.size()+1);
						myv.id = tsreq.getRequestorId();
						myv.msg = tsreq.getMessage();
						myv.file = tsreq.getFileName();
						myv.vector[myv.id - 1] = tsreq.getTimeStamp();
						myv.vector[tsreq.getForwarderId() - 1] = tsreq
								.getForwarderTimeStamp();
						myv.vector[id - 1] = logicalClock;
						if (!mapMap.containsKey(tsreq.getFileName())) {
							mapMap.put(myv.file,
									new HashMap<String, MyVector>());
						}
						mapMap.get(myv.file).put(key, myv);
					} else {
						System.out.println("map have key");
						if (!mapMap.containsKey(tsreq.getFileName())) {
							mapMap.put(tsreq.getFileName(),
									new HashMap<String, MyVector>());
						}
						MyVector myv = mapMap.get(tsreq.getFileName()).get(key);
						myv.key = key;
						myv.msg = tsreq.getMessage();
						myv.file = tsreq.getFileName();
						myv.vector[tsreq.getForwarderId() - 1] = tsreq
								.getForwarderTimeStamp();
						int flag = 1;
						for (int i : myv.vector) {
							flag *= i;
						}
						if (flag != 0) {
							if (queueMap.containsKey(myv.file)) {
								queueMap.get(myv.file).add(myv);
							} else {
								queueMap.put(myv.file,
										new PriorityBlockingQueue<MyVector>());
								queueMap.get(myv.file).add(myv);
							}
						}
					}
					pw.println("Bye.");
					System.out.println("The request reached here-"
							+ tsreq.getMessage());

					break;
				case TAILSERVER_WRITE_RESPONSE:

					synchronized (this) {
						pw.println("Bye.");
						writeFile(localReqHistoryMap.get(req.getFileName())
								.get(req.getMessage()));
						if (id != 1) {
							client.makeRequest("localhost", listenPort - 1, req);
						}
					}
					break;
				case TAILSERVER_READ:
					synchronized (this) {
						String line = readFile(req.getFileName());
						pw.println(line);
						pw.println("Bye.");
					}
					break;
				}

			}

			pw.close();
			os.close();
			br.close();

		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	private TailServerRequest parseTailServerRequest(String msg) {
		TailServerRequest tsreq = new TailServerRequest();
		String[] temp = msg.split(":");
		if (temp[0].endsWith("WRITE")) {
			tsreq.setType(RequestType.TAILSERVER_WRITE);
		} else {
			tsreq.setType(RequestType.TAILSERVER_READ);
		}
		tsreq.setFileName(temp[1]);
		tsreq.setMessage(temp[2]);
		tsreq.setTimeStamp(Integer.parseInt(temp[3]));
		tsreq.setSeqNum(Integer.parseInt(temp[4]));
		tsreq.setRequestorId(Integer.parseInt(temp[5]));
		tsreq.setForwarderId(Integer.parseInt(temp[6]));
		tsreq.setForwarderTimeStamp(Integer.parseInt(temp[7]));
		return tsreq;
	}

	private TailServerRequest getTailServerRequest(ServerRequest sreq) {
		TailServerRequest tsreq = new TailServerRequest();
		tsreq.setFileName(sreq.getFileName());
		tsreq.setForwarderId(id);
		tsreq.setForwarderTimeStamp(logicalClock);
		tsreq.setMessage(sreq.getMessage());
		tsreq.setRequestorId(sreq.getRequestorId());
		tsreq.setSeqNum(sreq.getSeqNum());
		tsreq.setTimeStamp(sreq.getTimeStamp());
		tsreq.setType(RequestType.TAILSERVER_WRITE);
		return tsreq;
	}

	private ServerRequest parseServerRequest(String msg) {
		ServerRequest sreq = new ServerRequest();
		String[] temp = msg.split(":");
		if (temp[0].endsWith("WRITE")) {
			sreq.setType(RequestType.SERVER_WRITE);
		} else {
			sreq.setType(RequestType.SERVER_READ);
		}
		sreq.setFileName(temp[1]);
		sreq.setMessage(temp[2]);
		sreq.setTimeStamp(Integer.parseInt(temp[3]));
		sreq.setSeqNum(Integer.parseInt(temp[4]));
		sreq.setRequestorId(Integer.parseInt(temp[5]));
		return sreq;
	}

	protected Request parseRequest(String incoming) {
		String[] temp = incoming.split(":");
		Request r = new Request();
		if (temp[0].equals("CLIENT_READ")) {
			r.setType(RequestType.CLIENT_READ);
		} else if (temp[0].equals("CLIENT_WRITE")) {
			r.setType(RequestType.CLIENT_WRITE);
		} else if (temp[0].equals("SERVER_READ")) {
			r.setType(RequestType.SERVER_READ);
		} else if (temp[0].equals("SERVER_WRITE")) {
			r.setType(RequestType.SERVER_WRITE);
		} else if (temp[0].equals("TAILSERVER_WRITE")) {
			r.setType(RequestType.TAILSERVER_WRITE);
		} else if (temp[0].equals("TAILSERVER_READ")) {
			r.setType(RequestType.TAILSERVER_READ);
		} else if (temp[0].equals("TAILSERVER_WRITE_RESPONSE")) {
			r.setType(RequestType.TAILSERVER_WRITE_RESPONSE);
		} else {
			r.setType(RequestType.TAILSERVER_READ_RESPONSE);
		}
		r.setFileName(temp[1]);
		r.setMessage(temp[2]);
		return r;
	}

	public boolean stopServer() {
		stop = true;
		return true;
	}

	private synchronized void writeFile(Request req) {
		File f = new File(req.getFileName());
		FileWriter fw = null;
		try {
			fw = new FileWriter(f, true);
			fw.write(req.getMessage() + "\n");
			fw.flush();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				fw.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	private synchronized String readFile(String fileName) {
		File readFile = new File(fileName);
		BufferedReader br;
		String lastLine = null;
		try {
			br = new BufferedReader(new InputStreamReader(new FileInputStream(
					readFile)));

			String temp1 = null;
			while ((temp1 = br.readLine()) != null) {
				lastLine = temp1;
			}
			br.close();
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return lastLine;
	}

	public static void main(String[] args) {
		MyServer serv = new MyServer("config.properties");
		serv.startServer();
	}
}

class Client {

	public String makeRequest(String host, int port, Request req) {
		String resp = null;
		try {
			Socket socket = new Socket(host, port);
			OutputStream os = socket.getOutputStream();
			PrintWriter pw = new PrintWriter(os, true);
			BufferedReader br = new BufferedReader(new InputStreamReader(
					socket.getInputStream()));
			pw.println(req.toString());
			pw.flush();
			String res = null;
			while ((res = br.readLine()) != null) {
				if (res.equals("Bye.")) {
					break;
				} else {
					resp = res;
				}
			}

			pw.close();
			os.close();
			br.close();

		} catch (UnknownHostException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}

		return resp;
	}
}

class MyVector implements Comparable<MyVector> {
	int[] vector;
	int id;
	boolean isready = false;
	String key;
	String msg;
	String file;

	public MyVector(int size) {
		vector = new int[size];
	}

	@Override
	public int compareTo(MyVector arg0) {
		boolean sm_or_eq = true;
		boolean lg_or_eq = true;
		for (int i = 0; i < vector.length; i++) {
			sm_or_eq = sm_or_eq && (vector[i] <= arg0.vector[i]);
			if (!sm_or_eq)
				break;
		}
		if (sm_or_eq) {
			for (int i = 0; i < vector.length; i++) {
				if (vector[i] < arg0.vector[i])
					return -1;
			}
		} else {
			for (int i = 0; i < vector.length; i++) {
				lg_or_eq = lg_or_eq && (vector[i] >= arg0.vector[i]);
				if (!lg_or_eq)
					break;
			}
		}
		if (lg_or_eq) {
			for (int i = 0; i < vector.length; i++) {
				if (vector[i] > arg0.vector[i])
					return 1;
			}
		}

		if (id < arg0.id) {
			return -1;
		} else {
			return 1;
		}
	}
}